package com.king.ml.app

import com.king.ml.annotation.{Runner, Task}
import com.king.ml.base.{BaseRunner, BaseTask}
import com.king.ml.enums.{PathEnum, TaskNameEnum}
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.serializer.KryoSerializer
import org.apache.spark.sql.SparkSession
import org.springframework.core.io.support.{PathMatchingResourcePatternResolver, ResourcePatternResolver}

import scala.reflect.runtime.universe
import scala.reflect.runtime.universe.typeOf


object FeatureContextApp {


  def main(args: Array[String]): Unit = {

    require(args.length == 1, "task name")
    val condition = TaskNameEnum.getEnumType(args.head)
    val (task: BaseTask, runner: BaseRunner) = scanTaskrunner(condition)

    val sparkConf = task.initConf()
      .setAppName(getClass.getSimpleName + condition)
      .set("spark.serializer", classOf[KryoSerializer].getName)
      .set("dfs.replication", "2")
      .set("dfs.client.block.write.locateFollowingBlock.retries", "15")

    sparkConf.set("task.runner", runner.getClass.getCanonicalName)
    implicit val spark: SparkSession = SparkSession.builder
      .appName(condition.toString)
      .config(sparkConf)
      .enableHiveSupport()
      .getOrCreate()
    val checkpointDir = PathEnum.HDFS_TEMP + "/checkpoint/" + runner.getClass.getCanonicalName
    spark.sparkContext.setCheckpointDir(checkpointDir)
    val path = new Path(checkpointDir)
    val fileSystem = FileSystem.get(spark.sparkContext.hadoopConfiguration)
    fileSystem.deleteOnExit(path)
    task.around

  }

  private def scanTaskrunner(condition: TaskNameEnum.Value): (BaseTask, BaseRunner) = {
    val path = ResourcePatternResolver.CLASSPATH_ALL_URL_PREFIX +
      getClass.getPackage.getName.replaceAll("\\.", "/") + "/**/*.class"
    val resourcePatternResolver = new PathMatchingResourcePatternResolver()
    val resources = resourcePatternResolver.getResources(path)

    val taskRunners = resources.filter(x => !x.getFilename.contains("$")).map {
      resource =>
        val canonicalPath = resource.getURL.getPath
        // val subPath = canonicalPath.substring(canonicalPath.indexOf("classes") + 8)
        val subPath = canonicalPath.substring(canonicalPath.indexOf(".jar!") + 6)
        val qualifier = subPath.substring(0, subPath.indexOf(".class")).replace("/", ".")
        val clazz = Class.forName(qualifier)
        val annotations = universe.runtimeMirror(clazz.getClassLoader)
          .classSymbol(clazz)
          .annotations
        val optionTask = if (annotations.exists(_.tree.tpe == typeOf[Task])) Some(clazz.newInstance().asInstanceOf[BaseTask]) else None
        val optionRunner = if (annotations.exists(x => x.tree.tpe == typeOf[Runner])) Some(clazz.newInstance().asInstanceOf[BaseRunner]) else None
        optionTask -> optionRunner
    }
    val tasks = taskRunners.collect { case (Some(t), _) if t.taskName == condition => t }
    val runners = taskRunners.collect { case (_, Some(r)) if r.taskName == condition => r }
    if (tasks.length == 1 && runners.length == 1) return tasks.head -> runners.head
    throw new RuntimeException(s"扫描出${tasks.length}个task，${runners.length}个runner！")
  }


}
